package com.asiainfo.dacp.datastash.broker.facade;

import com.asiainfo.dacp.core.BeanFactory;
import com.asiainfo.dacp.datastash.broker.core.BrokerContext;
import com.asiainfo.dacp.datastash.broker.core.BrokerException;
import com.asiainfo.dacp.datastash.broker.core.StashUnitModel;
import com.asiainfo.dacp.datastash.broker.core.StashUnitProcessModel;
import com.asiainfo.dacp.datastash.broker.core.enums.UnitStateEnum;
import com.asiainfo.dacp.datastash.broker.core.enums.UnitTaskStateEnum;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnit;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnitDefine;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnitProcess;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnitTask;
import com.asiainfo.dacp.datastash.broker.facade.service.UnitCompileService;
import com.asiainfo.dacp.datastash.broker.facade.service.UnitLoaderService;
import com.asiainfo.dacp.datastash.broker.facade.state.UnitStateLogService;
import com.asiainfo.dacp.datastash.broker.facade.state.UnitTaskStateLogService;
import com.asiainfo.dacp.datastash.broker.tracelog.StashTraceLogVo;
import com.asiainfo.dacp.datastash.broker.tracelog.StashTraceLogger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.text.ParseException;
import com.asiainfo.dacp.datastash.domain.unit.repos.DataStashUnitRepository;
import com.google.common.base.Preconditions;

import java.util.List;
import java.util.Map;

/**
 * unit的开始处理的门面
 *
 * 1.当unit依赖到达时，进入该类完成处理状态对象的构建，完成相关状态的维护
 *
 * 2.完成unit的分片，调用dispatch完成任务规划
 *
 * 3.把规划完成的分片对象放入相应的队列中完成任务的启动分配
 *
 * @author MeiKefu
 * @date 2017-11-21
 */
@Service
public class UnitProcessFacade {

	@Autowired
	private StashTraceLogger tracelog;
	private String clzName = this.getClass().getSimpleName();

	@Autowired
	private UnitStateLogService unitStateLogService;

	@Autowired
	private UnitTaskStateLogService unitTaskStateLogService;

	@Autowired
	private StashUnitProcessModel stashUnitProcessModel;

	@Autowired
	private UnitCompileService unitCompileService;

	@Autowired
	private StashUnitModel stashUnitModel;

	@Autowired
	private DataStashUnitRepository dataStashUnitRepository;

	public void processUnit(StashUnit stashUnit) {

		// 记录redis状态
		stashUnit.running();

		try {
			unitStateLogService.running(stashUnit);
		} catch (BrokerException e1) {
			e1.printStackTrace();
			tracelog.error(StashTraceLogVo.buildStashTraceLogVo(stashUnit, clzName, "error::", "状态更新失败"));
			return;
		}

		tracelog.info(
				StashTraceLogVo.buildStashTraceLogVo(stashUnit, clzName, "process:start", "构建处理对象stashProcessing"));

		StashUnitDefine stashUnitDefine = BrokerContext.getInstance().getStashUnitDefine(stashUnit.getUnitCode());

		if (stashUnitDefine == null) {
			tracelog.error(StashTraceLogVo.buildStashTraceLogVo(stashUnit, clzName, "fetchStashUnit:error",
					"获取静态的StashUnit配置失败"));
			return;
		}
		StashUnitProcess stashUnitProcess = unitCompileService.compile(stashUnit);

		if (stashUnitProcess != null) {
			try {

				// TODO 注意事务

				if (stashUnitProcess.getTasks() == null || stashUnitProcess.getTasks().size() == 0) {
					tracelog.error(StashTraceLogVo.buildStashTraceLogVo(stashUnit, clzName, "process:compile:error",
							"编译Unit错误，子任务数为空"));
					try {
						stashUnit.failed();
						unitStateLogService.update(stashUnit, UnitStateEnum.FAIL, "编译Unit错误，子任务数为空");

					} catch (BrokerException e) {
						e.printStackTrace();
					}
					return;
				}

				// 记录状态表
				unitStateLogService.flowlog(stashUnit, UnitStateEnum.RUNNING_COMPILE,
						"编译成功， 任务数:: " + stashUnitProcess.getTasks().size());

				unitTaskStateLogService.create(stashUnitProcess.getTasks());
				processTasks(stashUnitProcess);
			} catch (Exception e) {
				e.printStackTrace();
			}

		} else {

			tracelog.error(StashTraceLogVo.buildStashTraceLogVo(stashUnit, clzName, "process:build:error",
					"编译Unit错误，构建unitProcess对象为空"));
			try {
				stashUnit.failed();
				unitStateLogService.update(stashUnit, UnitStateEnum.FAIL, "编译Unit错误，构建unitProcess对象为空");
			} catch (BrokerException e) {
				e.printStackTrace();
			}
		}
	}

	private void processTasks(StashUnitProcess stashUnitProcess) {

		// 放入procsslist
		stashUnitProcessModel.pushUnitProcessList(stashUnitProcess);

		// 放入taskWaitQueue
		List<StashUnitTask> tasks = stashUnitProcess.getTasks();
		stashUnitProcessModel.pushUnitTaskWaitQueue(tasks);

		// 更新task状态
		for (StashUnitTask task : tasks) {
			stashUnitProcessModel.updateUnitTaskState(task.getTaskProcId(), UnitTaskStateEnum.BLOCKED_RELAY.value());
			unitTaskStateLogService.update(task, UnitTaskStateEnum.BLOCKED_RELAY);
		}

	}

	public void checkUnitResult(StashUnitTask stashUnitTask) {
		StashUnitProcess stashUnitProcess = stashUnitProcessModel.getUnitProcessList(stashUnitTask.getUnitProcId());

		List<StashUnitTask> tasks = stashUnitProcess.getTasks();

		Preconditions.checkArgument(tasks != null && tasks.size() > 0, "任务数为0，检查redis状态是否正常. ");

		boolean over = true;
		StashUnitModel stashUnitModel = BeanFactory.getBean(StashUnitModel.class);
		StashUnit stashUnit = stashUnitModel.getStashUnit(stashUnitTask.getUnitProcId());
		try {

			for (StashUnitTask task : tasks) {
				String taskState = stashUnitProcessModel.getUnitTaskState(task.getTaskProcId());

				Preconditions.checkNotNull(taskState);

				if (UnitTaskStateEnum.FAIL.value().equalsIgnoreCase(taskState)) {
					//
					stashUnit.failed();
					unitStateLogService.update(stashUnit, UnitStateEnum.FAIL,
							"分片任务" + stashUnitTask.getTaskProcId() + "执行失败");
					unitStateLogService.fail(stashUnit);

					over = false;
					break;
				}

				if (!UnitTaskStateEnum.SUCESS.value().equalsIgnoreCase(taskState)) {
					over = false;
					break;
				}
			}
			if (over) {
				unitSuccess(stashUnit, stashUnitTask);
			}
		} catch (BrokerException e) {
			e.printStackTrace();
		}
	}

	public void unitSuccess(StashUnit stashUnit, StashUnitTask stashUnitTask) throws BrokerException {
		stashUnit.success();
		unitStateLogService.update(stashUnit, UnitStateEnum.SUCESS, "执行成功");

		Map<String, StashUnitDefine> stashUnitDefs = BrokerContext.getInstance().getStashUnitDefs();
		for (StashUnitDefine stashUnitDefine : stashUnitDefs.values()) {
			if (stashUnitDefine.getUnitCode().equalsIgnoreCase(stashUnitTask.getUnitCode())) {
				try {
					// stashUnitModel.delUnit(stashUnit.getUnitProcId());
					if (stashUnit.isTurnover()) {
						String nextBatchNo = UnitLoaderService.lastBatchNo(stashUnitTask.getBatchNo(),
								stashUnitDefine.getBatchType(), stashUnitDefine.getBatchStep());
						stashUnitModel.updateUnitBatchNo(stashUnitDefine.getUnitCode(), nextBatchNo);
						dataStashUnitRepository.updateBatchNo(stashUnitDefine.getUnitCode(), nextBatchNo);
					}
					//
					unitStateLogService.over(stashUnit);
				} catch (ParseException e) {
					e.printStackTrace();
				}
			}
		}
	}

}
